package cn.piflow.rpc

import cn.piflow.local.SparkFlowEngine
import cn.piflow.{FlowEngine, JobService, StatService}
import org.apache.spark.sql.SparkSession

object FlowEngineServer {
	def main(args: Array[String]) {
		val httpPort = if (args.length > 0) {
			args(0).toInt
		} else {
			1224
		};
		val httpServletPath = if (args.length > 1) {
			args(1)
		} else {
			"/rpc"
		};

		new FlowEngineServer().start(httpPort, httpServletPath);
	}
}

class FlowEngineServer {
	var engine: SparkFlowEngine = null;

	def start(httpPort: Int, httpServletPath: String) = {
		val spark = SparkSession.builder.master("local[4]")
			.getOrCreate();
		engine = new SparkFlowEngine(spark);
		engine.start();

		val rcls: RemoteClassLoaderServiceImpl = new RemoteClassLoaderServiceImpl();

		val server = new HttpRpcServer();
		//set default class loader, which will load remote class generated by Shell
		server.serializer.setDefaultClassLoader(rcls.getClassLoader(this.getClass.getClassLoader));

		server.serve(classOf[JobService], engine.getJobService()).
			serve(classOf[StatService], engine.getStatService()).
			serve(classOf[EchoService], new EchoServiceImpl()).
			serve(classOf[RemoteClassLoaderService], rcls).
			start(httpPort, httpServletPath);
	}

	def shutdown() = {
		if (engine != null)
			engine.shutdown();
	}
}

class RemoteFlowEngine(httpServletUrl: String) extends FlowEngine {
	val rpcClient = new HttpRpcClient(httpServletUrl);
	lazy val proxyEchoService: EchoService = rpcClient.createProxy(classOf[EchoService].getName, classOf[EchoService]);
	lazy val proxyJobService: JobService = rpcClient.createProxy(classOf[JobService].getName, classOf[JobService]);
	lazy val proxyStatService: StatService = rpcClient.createProxy(classOf[StatService].getName, classOf[StatService]);
	lazy val proxyClassLoaderService: RemoteClassLoaderService = rpcClient.createProxy(classOf[RemoteClassLoaderService].getName, classOf[RemoteClassLoaderService]);

	def getJobService(): JobService = proxyJobService;

	//echo, for test
	def getEchoService(): EchoService = proxyEchoService;

	def getStatService(): StatService = proxyStatService;

	def registerRemoteClassLoaderServer(classLoaderServiceUrl: String): Unit = {
		proxyClassLoaderService.registerRemoteClassLoaderServer(classLoaderServiceUrl);
	}
}

/**
	* echo what it receives
	*/
trait EchoService {
	def echo(params: Any*): Seq[_];
}

class EchoServiceImpl extends EchoService {
	def echo(params: Any*): Seq[_] = {
		params.foreach(println(_));
		params;
	};
}